/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.example.ordermessage;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static org.apache.rocketmq.example.quickstart.Producer.DEFAULT_NAMESRVADDR;

/**
 * 顺序消息
 */

public class Producer {
    public static final String TOPIC = "TopicTestOrder";
    public static void main(String[] args) throws MQClientException {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
            producer.setNamesrvAddr(DEFAULT_NAMESRVADDR);
            producer.start();

            /**
             * 注意理解：
             * TAG：是用来过滤消息的
             *
             */


            List<Order> orderList = getOrderList();

            for (int i = 0; i < orderList.size(); i++) {

                Order order = orderList.get(i);
                String s = JSON.toJSONString(order);
                Message msg =
                        new Message(TOPIC, "*"
                                , "KEY" + i, s.getBytes(StandardCharsets.UTF_8));

                // 想要保证消息顺序，发送消息时候要保证消息发送到指定的队列中
                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Long id = (Long) arg;

                        int index = id.intValue() % mqs.size();
                        // 选择具体的消息队列
                        System.out.println("index = " + index + ", mqs = " + new String(msg.getBody()));
                        return mqs.get(index);
                    }
                }, order.getOrderId());
            }

            TimeUnit.SECONDS.sleep(3);

            producer.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
            throw new MQClientException(e.getMessage(), null);
        }
    }


  public static   List<Order> getOrderList() {

        List<Order> list = new ArrayList<>();
        Order order1 = new Order();
        order1.setOrderId(10001l);
        order1.setDesc("创建订单1");
        list.add(order1);

        Order order2 = new Order();
        order2.setOrderId(10002l);
        order2.setDesc("创建订单2");
        list.add(order2);


      Order order3 = new Order();
      order3.setOrderId(10003l);
      order3.setDesc("创建订单3");
      list.add(order3);



      Order payOrder1 = new Order();
      payOrder1.setOrderId(10001l);
      payOrder1.setDesc("付款订单1");
      list.add(payOrder1);


      Order payOrder2 = new Order();
        payOrder2.setOrderId(10002l);
        payOrder2.setDesc("付款订单2");
        list.add(payOrder2);



        Order payOrder3 = new Order();
        payOrder3.setOrderId(10003l);
        payOrder3.setDesc("付款订单3");
        list.add(payOrder3);

      Order settleOrder1 = new Order();
      settleOrder1.setOrderId(10001l);
      settleOrder1.setDesc("结算订单1");
      list.add(settleOrder1);

      Order settleOrder2 = new Order();
      settleOrder2.setOrderId(10002l);
      settleOrder2.setDesc("结算订单2");
      list.add(settleOrder2);


        Order settleOrder3 = new Order();
        settleOrder3.setOrderId(10003l);
        settleOrder3.setDesc("结算订单3");
        list.add(settleOrder3);


      return list;
    }
}
